package com.k2data.service;

import k2paltform.common.util.*;
import org.apache.avro.Schema;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;

public class SchemaService implements IService {
	public static final String SENSOR_CYCLE_LIST = "cycle.list";

	public static final Logger LOG = LoggerFactory.getLogger(SchemaService.class);

	public static final String HEALTH_CHECK_SCHEMA_ID = "kmx_test_schema";
	public static final String HEALTH_CHECK_SCHEMA = "{\"type\":\"record\",\"name\":\"kmx_test_schema\",\"namespace\":\"test.dataplatform\",\"aliases\":[\"kmx_test_schema\"]," +
			"\"fields\":" +
			"[" +
			"{\"name\":\"fieldgroupsysid\",\"type\":\"string\"}," +
			"{\"name\":\"timestamp\",\"type\":\"long\"}," +
			"{\"name\":\"i_deviceid\",\"type\":\"string\", \"iskey\":true, \"aliases\":[\"deviceid\"]}" +
			"]}";
	public static final String HEALTH_CHECK_ASSET_ID = "{\"deviceid\":\"HEALTH_CHECK\"}";

	private ZkSchema schemas;
	private ZookeeperService zkService;

	private ServiceState state;

	public SchemaService(String zookeeperUrl) throws Exception {
		this(zookeeperUrl, Constants.ZOOKEEPER_NAMESPACE);
	}

	public SchemaService(Configuration conf) throws Exception {
		this(conf.getString(ParamNames.ZOOKEEPER_URL), conf.getString(ParamNames.ZOOKEEPER_NAMESPACE, Constants.ZOOKEEPER_NAMESPACE));
	}

	public SchemaService(String zookeeperUrl, String namespace) throws Exception {
		if(zookeeperUrl == null || zookeeperUrl.length() == 0 || namespace == null || namespace.length() == 0) {
			LOG.error("Can't find zookeeper configuration!");
			return;
		}

		Configuration conf = new Configuration();
		conf.loadFromZookeeper(zookeeperUrl, namespace);
		zkService = new ZookeeperService(conf.getString(ParamNames.ZOOKEEPER_URL, zookeeperUrl), conf.getString(ParamNames.ZOOKEEPER_NAMESPACE, namespace));
		schemas = new ZkSchema(Constants.ZOOKEEPER_SCHEMA_PATH);
		state = ServiceState.INITED;
	}

	public boolean isServiceAvail() {
		if(state == ServiceState.STARTED) {
			return true;
		}
		else {
			return false;
		}
	}

	public void checkServiceAvail() {
		if(!isServiceAvail()) {
			LOG.error("Can't use service when it's state is " + this.state);
			throw new IllegalStateException("Can't use service when it's state is " + this.state);
		}
	}

	public boolean checkHealth() {
		if(!isServiceAvail()) {
			LOG.error("service is not available. Health Check fails.");
			return false;
		}

		if(!schemas.getSchemas().containsKey(HEALTH_CHECK_SCHEMA_ID)) {
			LOG.error("Can't find Health Check Schema!");
			return false;
		}
		return true;
	}

	public Schema getSchema(String schemaId) {
		checkServiceAvail();
		if(schemaId == null || schemaId.length() == 0) {
			LOG.error("Can't get schema for schemaId: " + schemaId + ".");
			return null;
		}
		return schemas.getSchema(schemaId);
	}

	public String[] getIdFieldsBySchemaId(String schemaId) {
		checkServiceAvail();
		if(schemaId == null || schemaId.length() == 0) {
			LOG.error("Can't get id group for schemaId: " + schemaId + ".");
			return null;
		}
		return schemas.getIdFields(schemaId);
	}

	public String[] getIdFieldsByFieldGroupId(String fieldGroupId) {
		checkServiceAvail();
		if(fieldGroupId == null || fieldGroupId.length() == 0) {
			LOG.error("Can't get id group for fieldGroupId: " + fieldGroupId + ".");
			return null;
		}

		String schemaId = getSchemaId(fieldGroupId);
		if(schemaId == null || schemaId.length() == 0) {
			LOG.error("Can't get schemaId with fieldGroupId: " + fieldGroupId + ".");
			return null;
		}
		return schemas.getIdFields(schemaId);
	}

	public FieldGroup getFieldGroup(String schemaId) {
		checkServiceAvail();
		if(schemaId == null || schemaId.length() == 0) {
			LOG.error("Can't get field group for schemaId: " + schemaId + ".");
			return null;
		}
		return schemas.getFieldGroup(schemaId);
	}

	public Map<String, Schema> getSchemas() {
		checkServiceAvail();
		return schemas.getSchemas();
	}

	public Map<String, FieldGroup> getFieldGroups() {
		checkServiceAvail();
		return schemas.getFieldGroups();
	}

	public Map<String, String[]> getIdFields() {
		checkServiceAvail();
		return schemas.getIdFields();
	}

	public static Map<String, Schema> getSchemasWithoutWatcher(String zookeeperUrl) throws Exception {
		if(zookeeperUrl == null || zookeeperUrl.length() == 0) {
			LOG.error("Can't find zookeeper url configuration!");
			return null;
		}
		return getSchemasWithoutWatcher(zookeeperUrl,Constants.ZOOKEEPER_NAMESPACE);
	}

	public static Map<String, Schema> getSchemasWithoutWatcher(Configuration conf) throws Exception {
		if(conf == null || conf.getString(ParamNames.ZOOKEEPER_URL) == null) {
			LOG.error("Can't find zookeeper url configuration!");
			return null;
		}
		return getSchemasWithoutWatcher(conf.getString(ParamNames.ZOOKEEPER_URL), conf.getString(ParamNames.ZOOKEEPER_NAMESPACE, Constants.ZOOKEEPER_NAMESPACE));
	}

	public static Map<String, Schema> getSchemasWithoutWatcher(String zookeeperUrl, String namespace) throws Exception {
		if(zookeeperUrl == null || zookeeperUrl.length() == 0 || namespace == null || namespace.length() == 0) {
			LOG.error("Can't find zookeeper configuration!");
			return null;
		}
		ZookeeperService tmpZkService = new ZookeeperService(zookeeperUrl, namespace);
		tmpZkService.start();
		if(!tmpZkService.checkExists(Constants.ZOOKEEPER_SCHEMA_PATH))
			return null;

		ZkSchema ret = new ZkSchema();
		ret.load(tmpZkService.getChildrenData(Constants.ZOOKEEPER_SCHEMA_PATH));
		tmpZkService.close();
		return ret.getSchemas();
	}

	public ZookeeperService getZkService() {
		return this.zkService;
	}

	public long[] getSensorCycles(String schemaId, String sensorId) {
		checkServiceAvail();
		if(schemaId == null || schemaId.length() == 0 || sensorId == null || sensorId.length() == 0) {
			LOG.error("Can't get sensor cycles for schemaId: " + schemaId + ", sensorId: " + sensorId);
			return null;
		}

		Schema.Field sensorField = getSensorField(schemaId, sensorId);
		if(sensorField != null) {
			String cycles = sensorField.getProp(SENSOR_CYCLE_LIST);
			if (cycles != null) {
				return CycleList.getCycleListFromStr(cycles);
			}
		}
		return null;
	}

	public Map<String, String> getSensorDataTypes(String fileGroupId, Set<String> sensorIds) {
		checkServiceAvail();
		if(fileGroupId == null || fileGroupId.length() == 0 || sensorIds == null || sensorIds.size() == 0) {
			LOG.error("Can't get sensor types for fieldGroupId: " + fileGroupId);
			return null;
		}

		Map<String, Schema> cachedSchemas = schemas.getSchemas();
		Schema schema = null;
		for(String schemaId : cachedSchemas.keySet()) {
			String namespace = cachedSchemas.get(schemaId).getNamespace();
			Set<String> aliases = cachedSchemas.get(schemaId).getAliases();
			if(aliases.size() < 1) {
				LOG.warn("FieldGroupId " + fileGroupId + " does not exist.");
				continue;
			}
			else {
				for(String aliase : aliases) {
					if(aliase.substring(namespace.length() + 1).equalsIgnoreCase(fileGroupId)) {
						schema = cachedSchemas.get(schemaId);
					}
				}
			}

			if(schema != null)
				break;
		}

		if(schema == null)
			return null;

		Map<String, String> sensorId2Type = new HashMap<>();
		for(Schema.Field field : schema.getFields()) {
			if(AvroUtil.isSensorField(field)) {
				String sensorId = null;
				for (String aliase : field.aliases()) {
					if (sensorIds.contains(aliase)) {
						sensorId = aliase;
						break;
					}
				}
				if (sensorId != null) {
					sensorId2Type.put(sensorId, getFieldType(field));
				}
			}
		}

		return sensorId2Type;
	}

	public Map<String, String> getAllSensorDataTypes(String schemaId) {
		checkServiceAvail();
		if(schemaId == null || schemaId.length() == 0) {
			LOG.error("Can't get sensor types for shemaId: " + schemaId);
			return null;
		}

		Schema schema = schemas.getSchema(schemaId);
		if(schema == null)
			return null;

		Map<String, String> sensorId2Type = new HashMap<>();
		for(Schema.Field field : schema.getFields()) {
			if(AvroUtil.isSensorField(field)) {
				String sensorId = null;
				for (String aliase : field.aliases()) {
					sensorId = aliase;
					break;
				}
				if (sensorId != null) {
					sensorId2Type.put(sensorId, getFieldType(field));
				}
			}
		}

		return sensorId2Type;
	}

	public Map<String, String> getIdFieldId2IdFieldSysId(String schemaId) {
		checkServiceAvail();
		if(schemaId == null || schemaId.length() == 0) {
			LOG.error("Can't get id field id map for schemaId: " + schemaId);
			return null;
		}

		Map<String, String> idFieldId2IdFieldSysId = new HashMap<>();
		Schema schema = schemas.getSchema(schemaId);
		if(schema == null)
			return null;

		for(Schema.Field field : schema.getFields()) {
			if(AvroUtil.isIdField(field)) {
				String idFieldId = null;
				for(String aliase : field.aliases()) {
					idFieldId = aliase;
					break;
				}
				if(idFieldId != null) {
					idFieldId2IdFieldSysId.put(idFieldId, field.name());
				}
			}
		}
		return idFieldId2IdFieldSysId;
	}

	public Map<String, String> getSensorId2SensorSysId(String schemaId) {
		checkServiceAvail();
		if(schemaId == null || schemaId.length() == 0) {
			LOG.error("Can't get sensor id map for schemaId: " + schemaId);
			return null;
		}

		Map<String, String> sensorId2SensorSysId = new HashMap<>();
		Schema schema = schemas.getSchema(schemaId);
		if(schema == null)
			return null;

		for(Schema.Field field : schema.getFields()) {
			if(AvroUtil.isSensorField(field)) {
				String sensorId = null;
				for(String aliase : field.aliases()) {
					sensorId = aliase;
					break;
				}
				if(sensorId != null) {
					sensorId2SensorSysId.put(sensorId, field.name());
				}
			}
		}
		return sensorId2SensorSysId;
	}

	private String getFieldType(Schema.Field field) {
		Schema fieldSchema = field.schema();
		if (fieldSchema.getName().equals("union")) {
			for (Schema scm : fieldSchema.getTypes()) {
				if (!scm.getName().equalsIgnoreCase("null")) {
					return scm.getName();
				}
			}
			return null;
		}
		else {
			return fieldSchema.getName();
		}
	}
	
	private Schema.Field getSensorField(String schemaId, String sensorId) {
		Schema schema = getSchema(schemaId);
		if(schema == null)
			return null;

		for(Schema.Field field : schema.getFields()) {
			if(AvroUtil.isSensorField(field)) {
				for (String aliase : field.aliases()) {
					if (aliase.equalsIgnoreCase(sensorId))
						return field;
				}
			}
		}
		return null;
	}

	
	public String getSchemaId(String fieldGroupId) {
		checkServiceAvail();
		if(fieldGroupId == null || fieldGroupId.length() == 0) {
			LOG.error("Can't get schemaId for fieldGroupId: " + fieldGroupId + ".");
			return null;
		}

		Map<String, Schema> cachedSchemas = schemas.getSchemas();
		for(String schemaId : cachedSchemas.keySet()) {
			String namespace = cachedSchemas.get(schemaId).getNamespace();
			Set<String> aliases = cachedSchemas.get(schemaId).getAliases();
			if(aliases.size() < 1) {
				LOG.warn("FieldGroupId id " + fieldGroupId + " does not exist.");
				continue;
			}
			else {
				for(String aliase : aliases) {
					if(aliase.substring(namespace.length() + 1).equalsIgnoreCase(fieldGroupId)) {
						return schemaId;
					}
					else {
						break;
					}
				}
			}
		}
		return null;
	}


	public String getFieldGroupId(String schemaId) {
		checkServiceAvail();
		if(schemaId == null || schemaId.length() == 0) {
			LOG.error("Can't get fieldGroupId for schemaId: " + schemaId + ".");
			return null;
		}

		Schema schema = schemas.getSchema(schemaId);
		if(schema != null) {
			String namespace = schema.getNamespace();
			Set<String> aliases = schema.getAliases();
			for(String aliase : aliases) {
				return aliase.substring(namespace.length() + 1);
			}
		}
		return null;
	}

	public List<String> getSensorIds(String schemaId) {
		checkServiceAvail();
		if(schemaId == null || schemaId.length() == 0) {
			LOG.error("Can't get sensorId list for schemaId: " + schemaId + ".");
			return null;
		}

		List<String> sensorIds = new ArrayList<>();
		Schema schema = getSchema(schemaId);
		if(schema == null)
			return null;

		for(Schema.Field field : schema.getFields()) {
			if(AvroUtil.isSensorField(field)) {
				if (field.aliases().size() < 1) {
					LOG.warn("There is no sensor id for sensor " + field.name());
					sensorIds.add("");
				} else {
					for (String aliase : field.aliases()) {
						sensorIds.add(aliase);
						break;
					}
				}
			}
		}
		return sensorIds;
	}

	//store schema objects from memory to zookeeper, if exists, will be deleted and created
	public void storeSchemas(Map<String, Schema> schemas) throws Exception {
		checkServiceAvail();
		if(schemas == null || schemas.size() == 0)
			return;

		Map<String, byte[]> datas = new HashMap<>();
		Map<String, byte[]> assetSchemaDatas = new HashMap<>();
		for(String schemaId : schemas.keySet()) {
			datas.put(schemaId, schemas.get(schemaId).toString().getBytes());
			assetSchemaDatas.put(schemaId, Long.toString(System.currentTimeMillis()).getBytes());
		}
		zkService.create(Constants.ZOOKEEPER_SCHEMA_PATH, datas);
		zkService.create(Constants.ZOOKEEPER_ASSET_PATH, assetSchemaDatas);
	}

	public boolean existSchema(String schemaId) throws Exception {
		checkServiceAvail();
		if(schemaId == null || schemaId.length() == 0) {
			LOG.error("Can't check schema exist or not for schemaId: " + schemaId + ".");
			return false;
		}
		return zkService.checkExists(Constants.ZOOKEEPER_SCHEMA_PATH + "/" + schemaId);
	}

	//remove schema objects from zookeeper
	public void removeSchemas(List<String> schemaIds) throws Exception {
		checkServiceAvail();
		if(schemaIds == null || schemaIds.size() == 0)
			return;

		for(String schemaId : schemaIds) {
			zkService.delete(Constants.ZOOKEEPER_SCHEMA_PATH + ZookeeperService.ZOOKEEPER_PATH_SEPARATOR + schemaId);
			zkService.delete(Constants.ZOOKEEPER_ASSET_PATH + ZookeeperService.ZOOKEEPER_PATH_SEPARATOR + schemaId);
		}
	}

	//insert some new schemas from memory to zookeeper
	public void updateSchemas(Map<String, Schema> schemas) throws Exception {
		checkServiceAvail();
		if(schemas == null || schemas.size() == 0)
			return;

		for(String schemaId : schemas.keySet()) {
			zkService.update(Constants.ZOOKEEPER_SCHEMA_PATH + ZookeeperService.ZOOKEEPER_PATH_SEPARATOR + schemaId, schemas.get(schemaId).toString().getBytes());
		}
	}

	//store schema objects from memory to zookeeper
	public void storeSchema(String schemaId, Schema schema) throws Exception {
		checkServiceAvail();
		if(schemaId == null || schemaId.length() == 0 || schema == null) {
			LOG.error("Can't store schema for schemaId: " + schemaId + ".");
			return;
		}

		zkService.create(Constants.ZOOKEEPER_SCHEMA_PATH + ZookeeperService.ZOOKEEPER_PATH_SEPARATOR + schemaId, schema.toString().getBytes());
		zkService.create(Constants.ZOOKEEPER_ASSET_PATH + ZookeeperService.ZOOKEEPER_PATH_SEPARATOR + schemaId, Long.toString(System.currentTimeMillis()).getBytes());
	}

	//remove schema objects from zookeeper
	public void removeSchema(String schemaId) throws Exception {
		checkServiceAvail();
		if(schemaId == null || schemaId.length() == 0) {
			LOG.error("Can't remove schema for schemaId: " + schemaId + ".");
			return;
		}
		zkService.delete(Constants.ZOOKEEPER_SCHEMA_PATH + ZookeeperService.ZOOKEEPER_PATH_SEPARATOR + schemaId);
		zkService.delete(Constants.ZOOKEEPER_ASSET_PATH + ZookeeperService.ZOOKEEPER_PATH_SEPARATOR + schemaId);
	}

	//insert some new schemas from memory to zookeeper
	public void updateSchema(String schemaId, Schema schema) throws Exception {
		checkServiceAvail();
		if(schemaId == null || schemaId.length() == 0 || schema == null) {
			LOG.error("Can't update schema for schemaId: " + schemaId + ".");
			return;
		}
		zkService.update(Constants.ZOOKEEPER_SCHEMA_PATH + ZookeeperService.ZOOKEEPER_PATH_SEPARATOR + schemaId, schema.toString().getBytes());
	}

	public void reload() throws Exception {
		checkServiceAvail();
		schemas.load(zkService.getChildrenData(Constants.ZOOKEEPER_SCHEMA_PATH));
	}

	@Override
	public void start() {
		if(zkService == null)
			return;

		zkService.start();

		String path = "";  //if root path does not exist, create it
		try {
			path = Constants.ZOOKEEPER_SCHEMA_PATH;
			if(!zkService.checkExists(path)) {
                zkService.create(path, Long.toString(System.currentTimeMillis()).getBytes());
				LOG.debug("Create schema path in Zookeeper.");
            }

			if(!zkService.checkExists(path + "/" + HEALTH_CHECK_SCHEMA_ID)) {
				zkService.create(path + "/" + HEALTH_CHECK_SCHEMA_ID, HEALTH_CHECK_SCHEMA.getBytes());  //create default schema for health check
				LOG.debug("Create schema health check node in Zookeeper.");
			}

			path = Constants.ZOOKEEPER_ASSET_PATH;
			if(!zkService.checkExists(path)) {
				zkService.create(path, Long.toString(System.currentTimeMillis()).getBytes());
				LOG.debug("Create asset path in Zookeeper.");
			}

			if(!zkService.checkExists(path + "/" + HEALTH_CHECK_SCHEMA_ID)) {
				zkService.create(
						Constants.ZOOKEEPER_ASSET_PATH + "/"
								+ HEALTH_CHECK_SCHEMA_ID + "/" + Constants.SCHEMA_ASSET_PREFIX
								+ AssetService.computeAssetSystemID(HEALTH_CHECK_ASSET_ID),
						HEALTH_CHECK_ASSET_ID.getBytes());
				LOG.debug("Create asset health check node in Zookeeper.");
			}

			if (!zkService.checkExists(Constants.ZOOKEEPER_SCHEMA_RELOAD_PATH)) {
				zkService.create(
						Constants.ZOOKEEPER_SCHEMA_RELOAD_PATH,
						String.valueOf(System.currentTimeMillis()).getBytes());
				LOG.info("Created schema zk reload node");
			}
		} catch (Exception e) {
			e.printStackTrace();
			LOG.error("Can't create " + path + " in Zookeeper.");
			throw new IllegalStateException("Can't start schema service.", e);
		}

		zkService.register(Constants.ZOOKEEPER_SCHEMA_PATH, schemas);
		try {
			schemas.load(zkService.getChildrenData(Constants.ZOOKEEPER_SCHEMA_PATH));
			state = ServiceState.STARTED;
		} catch (Exception e) {
			e.printStackTrace();
			zkService.unregister(Constants.ZOOKEEPER_SCHEMA_PATH, schemas);
			schemas.clearSchemas();
			LOG.error("Can't load schema data with watcher on " + Constants.ZOOKEEPER_SCHEMA_PATH + " path");
			throw new IllegalStateException("Can't start schema service.", e);
		}

		LOG.info("SchemaService is started.");
	}

	@Override
	public void close() {
		state = ServiceState.CLOSED;
		if(schemas != null)
			schemas.clearSchemas();
		if(zkService != null) {
			zkService.unregister(Constants.ZOOKEEPER_SCHEMA_PATH, schemas);
			zkService.close();
		}

		LOG.info("SchemaService is closed.");
	}
}
